package com.burges.net.dataStream.windows.windowAssigner

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.assigners._
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * Created by zhanglibing on 2019/2/17
  * 会话窗口的基本使用
  */
object SessionWindows {
  def main(args: Array[String]): Unit = {
    case class WebLog(id: Long, content: String, trackTime: Long)

    val parameterTool = ParameterTool.fromArgs(args)

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val inputStream = env.fromElements(WebLog(2, "flink", 21L), WebLog(4, "flink", 1L), WebLog(3, "flink", 4L))

    //通过EventTimeSessionWindows定义EventTime的会话窗口
//    inputStream.keyBy(_.id)
//      //
//      .window(EventTimeSessionWindows.withGap(Time.milliseconds(10)))
//      .process(...)
//
    //通过ProcessingTimeSessionWindows定义ProcessingTime的会话窗口
//    inputStream.keyBy(_.id)
//      //
//      .window(ProcessingTimeSessionWindows.withGap(Time.milliseconds(10)))
//      .process(...)
//
//    inputStream.keyBy(_.id).windowAll()

    /**
      * 通过动态的SessionGap来定义EventTime和ProcessTime SessionWindows
      */
    val eventTimeSessionWindows = inputStream.keyBy(_.id)
            //通过使用EventTimeSessionWindows定义EventTime滑动窗口
            .window(EventTimeSessionWindows.withDynamicGap(
              new SessionWindowTimeGapExtractor[WebLog] {
                override def extract(t: WebLog): Long = {
                  //动态指定并返回sessionGap
                  0L
                }
              }
            ))
//            .process(...)

    val processTimeSessionWindows = inputStream.keyBy(_.id)
            //通过使用ProcessingTimeSessionWindows定义ProcessTime滑动窗口
            .window(ProcessingTimeSessionWindows.withDynamicGap(
              new SessionWindowTimeGapExtractor[WebLog] {
                override def extract(t: WebLog): Long = {
                  //动态指定并返回sessionGap
                  0L
                }
              }
            ))
    //      .process(...)


  }




}
